package cn.com.sparkle.paxos.client;

import java.util.concurrent.atomic.AtomicInteger;

import cn.com.sparkle.paxos.client.deamon.MasterHeartBeatDeamon;
import cn.com.sparkle.paxos.event.EventsManager;
import cn.com.sparkle.paxos.iomodel.protobuffmessage.PaxosMessages.MessagePackage;
import cn.com.sparkle.paxos.net.NetNode;

public class PaxosClient {

	public static interface CommandCallBack {
		public void response(NetNode node, MessagePackage request);
	}

	private String[] senators;
	private CommandProcessor[] processor;
	private AtomicInteger flag = new AtomicInteger(0);
	private EventsManager eventsManager = new EventsManager();

	public PaxosClient(String[] senator, int tcpConnectNum,
			boolean isPromiseMaster) throws Exception {
		this.senators = senator;
		if (tcpConnectNum < 1) {
			tcpConnectNum = 1;
		}
		CommandProcessor.init(tcpConnectNum,eventsManager);
		processor = new CommandProcessor[tcpConnectNum];
		for (int i = 0; i < tcpConnectNum; ++i) {
			processor[i] = new CommandProcessor(senator, 5000, 1000,
					isPromiseMaster);
			processor[i].runDeamon();
		}

		MasterHeartBeatDeamon heartBeatDeamon = new MasterHeartBeatDeamon(
				processor);
		heartBeatDeamon.start();
	}
	

	public PaxosOperater getOperator() {
		int nextFlag;
		for (;;) {
			int curFlag = flag.get();
			if (curFlag == Integer.MAX_VALUE) {
				if (flag.compareAndSet(curFlag, 0)) {
					nextFlag = 0;
					break;
				}
			} else {
				nextFlag = curFlag + 1;
				if (flag.compareAndSet(curFlag, nextFlag)) {
					break;
				}
			}
		}
		return new PaxosOperater(processor[nextFlag % processor.length]);
	}
	
	public EventsManager getEventsManager() {
		return eventsManager;
	}


	public void changeSenator(String[] senator) throws Exception {
		for (CommandProcessor p : processor) {
			p.changeSenator(senator);
		}
	}
}
